package com.taobao.analyse.map;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import com.ibm.icu.util.Calendar;
import com.taobao.analyse.constants.SplitConstants;

public class MergeTimeResultMapper extends MapReduceBase implements
		Mapper<Writable, Text, Text, LongWritable> {

	@Override
	public void map(Writable key, Text value,
			OutputCollector<Text, LongWritable> output, Reporter reporter)
			throws IOException {
		String line = value.toString();
		String categoryName = "";
		String[] datas = line.split(SplitConstants.SPLIT_OUTPUT);
		//只分析数据格式为：12345	time@1_21#1_12
		if (datas.length < 2 || !datas[1].startsWith("time")) {
			return;
		}
		
		if(datas[1].startsWith("categoryName")){
			categoryName = datas[1].split(SplitConstants.SPLIT_AT)[1];
		}else{
			String[] timesInfo = datas[1].split(SplitConstants.SPLIT_AT);
			String[] timesStr = timesInfo[1].split(SplitConstants.SPLIT_POUND);
			for(String time : timesStr){
				String[] timeSplit = time.split(SplitConstants.SPLIT_UNDER_BAR);
				int dayOfWeekInt = Integer.parseInt(timeSplit[0]);
				String hour = timeSplit[1];
				String dayOfWeek = "";
				
				switch(dayOfWeekInt){
				case 1:
					dayOfWeek = "MONDAY";
					break;
				case 2:
					dayOfWeek = "TUESDAY";
					break;
				case 3:
					dayOfWeek = "WEDNESDAY";
					break;
				case 4:
					dayOfWeek = "THURSDAY";
					break;
				case 5:
					dayOfWeek = "FRIDAY";
					break;
				case 6:
					dayOfWeek = "SATURDAY";
					break;
				case 7:
					dayOfWeek = "SUNDAY";
					break;
				default:
					dayOfWeek = "error";
					return;
				}

				output.collect(new Text(datas[0]+SplitConstants.SPLIT_UNDER_BAR+categoryName+SplitConstants.SPLIT_UNDER_BAR+dayOfWeek), new LongWritable(1));
				output.collect(new Text(datas[0]+SplitConstants.SPLIT_UNDER_BAR+categoryName+SplitConstants.SPLIT_UNDER_BAR+hour),new LongWritable(1));
				output.collect(new Text(hour),new LongWritable(1));
				output.collect(new Text(dayOfWeek),new LongWritable(1));
				output.collect(new Text(dayOfWeek+SplitConstants.SPLIT_UNDER_BAR+hour),new LongWritable(1));
			}
		}
		
		
	}

}
